package com.hqq.rabbitmq.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitMQ相关配置类
 * @author huangqiqin
 * @date 2021/04/13 00:31
 **/
@Configuration
public class RabbitmqConfig {

    private static final Logger log = LoggerFactory.getLogger(RabbitmqConfig.class);

    /**
     * 注入 RabbitMQ 的链接工厂
     */
    @Autowired
    private CachingConnectionFactory connectionFactory;

    /**
     * 注入 RabbitMQ 消息监听器所在的容器工厂配置类
     */
    //@Autowired
    //private SimpleRabbitListenerContainerFactoryConfigurer containerFactoryConfigurer;

    /**
     * 创建单个消息者实例下的消息监听器的容器工厂
     * @return
     */
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory singleListenerContainerFactory(){

        // 创建消息监听器所有容器工厂
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        // 设置容器工厂所使用的容器
        factory.setConnectionFactory(connectionFactory);
        // 设置消息传输的格式（这里采用JSON格式）
        //factory.setMessageConverter(new Jackson2JsonMessageConverter());
        // 设置并发消费者实例的初始数量
        factory.setConcurrentConsumers(1);
        // 设置并发消费者实例的最大数量
        factory.setMaxConcurrentConsumers(1);
        // 设置并发消费者实例中每个实例拉取的消息数量
        factory.setPrefetchCount(1);

        return factory;
    }

    /**
     * 创建多个消息者实例下的消息监听器的容器工厂，主要针对高并发业务场景
     * @return
     */
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainerFactory(){

        // 创建消息监听器所有容器工厂
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        // 设置容器工厂所使用的容器
        factory.setConnectionFactory(connectionFactory);
        // 设置消息传输的格式（这里采用JSON格式）
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        // 设置并发消费者实例的初始数量
        factory.setConcurrentConsumers(10);
        // 设置并发消费者实例的最大数量
        factory.setMaxConcurrentConsumers(15);
        // 设置并发消费者实例中每个实例拉取的消息数量
        factory.setPrefetchCount(10);
        // 设置消息的确认消费模式。这里设置为NONE，表示不需要确认（MANUAL手动确认，AUTO自动确认）
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);

        return factory;
    }

    /**
     * 配置RabbitMQ发送消息模板
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate(){
        // 设置发送消息后进行确认
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 设置发送消息返回确认消息
        connectionFactory.setPublisherReturns(true);

        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);

        // 设置消息发送成功后的回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("==> 消息发送到RabbitMQ是否成功：{}，失败原因：{}。", ack, cause);
            }
        });

        // 设置消息发送失败后的回调
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replayCode, String replayText,
                                        String exchange, String routingKey) {
                log.error("==> 消息发送失败：replayCode={}, replayText={}, exchange={}, routingKey={}",
                        replayCode, replayText, exchange, routingKey);
            }
        });

        return rabbitTemplate;
    }

}
